package com.gy.flink.watermark;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;

/**
 * 每隔五秒计算最近10秒单词出现的次数
 */
public class TimeWindowWordCount {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);

        DataStreamSource<String> dataStrean = env.socketTextStream("localhost", 9999);

        dataStrean.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] split = value.split(",");
                for (String str : split) {
                    out.collect(Tuple2.of(str, 1));
                }
            }
        })
                .keyBy(0)
                .timeWindow(Time.seconds(10), Time.seconds(5))
//                .sum(1)
                .process(sumProcessWindow())
                .print()
                .setParallelism(1);

        env.execute(TimeWindowWordCount.class.getCanonicalName());

    }


    private static ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple, TimeWindow> sumProcessWindow() {
        /**
         * IN:输入数据类型
         * OUT:输出数据类型
         * KEY:分组的数据类型
         * W extends Window:窗口的类型
         */
        return new ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple, TimeWindow>() {

            @Override
            public void process(Tuple tuple, Context context, Iterable<Tuple2<String, Integer>> elements, Collector<Tuple2<String, Integer>> out) throws Exception {
                System.out.println("当前系统时间：" + LocalDateTime.now());
                System.out.println("window处理时间：" + LocalDateTime.ofInstant(Instant.ofEpochMilli(context.currentProcessingTime()), ZoneId.systemDefault()));
                System.out.println("window开始时间：" + LocalDateTime.ofInstant(Instant.ofEpochMilli(context.window().getStart()), ZoneId.systemDefault()));
                System.out.println("window结束时间：" + LocalDateTime.ofInstant(Instant.ofEpochMilli(context.window().getEnd()), ZoneId.systemDefault()));
                int count = 0;
                for (Tuple2<String, Integer> ele : elements) {
                    count++;
                }
                out.collect(Tuple2.of(tuple.getField(0), count));
            }
        };

    }

}
